package homework;

import com.shujia.utils.MyHBaseTools;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.TreeSet;

/*
    查询文科一班学生总分排名前10的学生（输出：学号，姓名，班级，总分）结果写到hbase
    students
    scores

    select t1.id,sum(score) as sum_score from (select * from students where clazz='文科一班') t1 join scores t2 on(t1.id=t2.id) group by t1.id order by sum_score desc limit 10;

    提示：要进行rowkey的合理设计
    三个设计原则：
        长度不宜过长
        散列性
        唯一性
   步骤1：定义一个学生类（学号，姓名，班级，总分），先将数据存储到HBase中
   步骤2：可以先将文科一班的学生过滤出来，放到集合（List）中
   步骤3：读取scores数据（边读取上面过滤的文科一班学生集合得到学号，然后根据学号使用过滤器查询scores表，计算出该学生的总分）
   步骤4：将每次计算的学生总成绩写入到一个集合中（TreeSet）姓名，班级，总分

 */
public class Test1 {
    public static void main(String[] args) throws Exception {
        /**
         *  步骤1：定义一个学生类（学号，姓名，班级，总分），先将数据存储到HBase中
         */
        //获取hbase的连接对象
        ArrayList<Object> objectArrayList = MyHBaseTools.getConnection();
        Connection conn = (Connection) objectArrayList.get(0);
        Admin admin = (Admin) objectArrayList.get(1);
        System.out.println("成功获取数据库连接对象: " + conn);
        System.out.println("成功获取数据库操作对象: " + admin);
        System.out.println("====================================================");
        //创建hbase表
        Boolean b1 = MyHBaseTools.createHbaseTable(admin, "students", "info");
        System.out.println(b1 ? "学生表创建成功！" : "学生表创建失败！！");
        Boolean b2 = MyHBaseTools.createHbaseTable(admin, "scores", "info");
        System.out.println(b2 ? "成绩表创建成功！" : "成绩表创建失败！！");
        //读取学生数据和成绩数据
        BufferedReader studentsInfoReader = new BufferedReader(new FileReader("hbase/data/students.csv"));
        BufferedReader scoresInfoReader = new BufferedReader(new FileReader("hbase/data/score.txt"));
        //调用方法将学生信息添加到hbase表中
        putStudentsInfo(conn, studentsInfoReader);
        //调用方法将成绩信息添加到hbase表中
        putScoresInfo(conn, scoresInfoReader);

        /**
         * 步骤2：可以先将文科一班的学生过滤出来，放到集合（List）中
         */
        List<Student> studentsInfoWithWenkeList = scanStudentsWithWenke(conn);

        /**
         * 步骤3：读取scores数据（边读取上面过滤的文科一班学生集合得到学号，然后根据学号使用过滤器查询scores表，计算出该学生的总分）
         */
        TreeSet<Student> studentTreeSet = getStudentsWithSumScore(conn, studentsInfoWithWenkeList);

        /**
         * 取出集合中的前10写回到hbase中
         */
        putSumScoreTop10Students(admin,conn,studentTreeSet);


    }

    public static void putSumScoreTop10Students(Admin admin, Connection conn, TreeSet<Student> studentTreeSet) throws Exception {
        //创建结果数据表
        MyHBaseTools.createHbaseTable(admin, "wenke1_sum_score_top10", "info");

        //获取表的实例
        Table wenke1SumScoreTop10 = MyHBaseTools.getTableExample(conn, "wenke1_sum_score_top10");

        //创建集合存储一行中其他列的Put对象
        ArrayList<Put> puts = new ArrayList<>();

        int i = 0;
        Put put = null;
        //遍历集合，取出前10个学生将信息添加到表中
        for (Student student : studentTreeSet) {
            if (i == 10) {
                break;
            }
            System.out.println(student.getName()+"已添加到数据库中...");
            byte[] rk = Bytes.toBytes(student.getId());
            put = new Put(rk);
            put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(student.getName()));
            put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("clazz"),Bytes.toBytes(student.getClazz()));
            put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("sumsore"),Bytes.toBytes(student.getSumScore()));

            puts.add(put);

            i++;
        }
        wenke1SumScoreTop10.put(puts);
    }

    /**
     * 获取每个学生的总成绩
     */
    public static TreeSet<Student> getStudentsWithSumScore(Connection conn, List<Student> studentsInfoWithWenkeList) throws Exception {
        //创建一个TreeSet集合，传入比较器，根据总成绩倒序排序
        TreeSet<Student> studentTreeSet = new TreeSet<>(new Comparator<Student>() {
            @Override
            public int compare(Student o1, Student o2) {
                //按照总分从高到底排序
                int i = o2.getSumScore() - o1.getSumScore();
                //总分一样，学号不一定一样
                int i2 = (i == 0) ? o2.getId().compareTo(o1.getId()) : i;
                return i2;
            }
        });

        //获取表的实例
        Table scores = MyHBaseTools.getTableExample(conn, "scores");
        //1500100001-100001

        /**
         * 步骤4：将每次计算的学生总成绩写入到一个集合中（TreeSet）姓名，班级，总分
         */
        //遍历传入的文科学生集合
        for (Student student : studentsInfoWithWenkeList) {
            String id = student.getId();
            //使用过滤期查询对应学号的学生信息
            Scan scan = new Scan();

            //创建PrefixFilter
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(id)); // 1500101000
            scan.setFilter(prefixFilter);
            //查询出每个学生的6门成绩
            ResultScanner scanner = scores.getScanner(scan);
            /*
             1500101000-1000001                   column=info:score, timestamp=1702605167802, value=100
             1500101000-1000002                   column=info:score, timestamp=1702605167802, value=78
             1500101000-1000003                   column=info:score, timestamp=1702605167802, value=81
             1500101000-1000007                   column=info:score, timestamp=1702605167802, value=5
             1500101000-1000008                   column=info:score, timestamp=1702605167802, value=87
             1500101000-1000009                   column=info:score, timestamp=1702605167802, value=28
             */
            int sum = 0;
            for (Result result : scanner) {
                int score = Integer.parseInt(Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("score"))));
                sum += score;
            }
            //将算出的总成绩赋值到学生总成绩属性中
            student.setSumScore(sum);
            //将计算好的学生对象添加到studentTreeSet集合中
            studentTreeSet.add(student);
        }
        return studentTreeSet;
    }

    public static List<Student> scanStudentsWithWenke(Connection conn) throws Exception {
        //获取表的实例
        Table students = MyHBaseTools.getTableExample(conn, "students");
        Scan scan = new Scan();

        //创建一个二进制比较器
        BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("文科一班"));

        //创建单列值过滤器
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"),
                Bytes.toBytes("clazz"),
                CompareOperator.EQUAL,
                binaryComparator);

        scan.setFilter(singleColumnValueFilter);

        //查询班级是文科一班的学生
        ResultScanner scanner = students.getScanner(scan);
        //创建一个集合存储查询到学生信息
        ArrayList<Student> list = new ArrayList<>();

        //遍历结果封装成学生对象添加到集合并返回
        //学生类（学号，姓名，班级，总分）
        for (Result result : scanner) {
            String id = Bytes.toString(result.getRow());
            //姓名
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            //班级
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            Student student = new Student(id, name, clazz, 0);
            list.add(student);
        }
        return list;

    }


    public static void putScoresInfo(Connection conn, BufferedReader bufferedReader) throws Exception {
        //获取表的实例
        Table scores = MyHBaseTools.getTableExample(conn, "scores");

        //创建一个集合，存储不同学生不同科目的成绩列
        ArrayList<Put> scoresPuts = new ArrayList<>();
        String line = null;
        Put put = null;
        while ((line = bufferedReader.readLine()) != null) {
            //1500100001,1000001,98
            String[] scoresInfo = line.split(",");
            //将学号和科目编号一起作为rowkey存储
            byte[] rk = Bytes.toBytes(scoresInfo[0] + "-" + scoresInfo[1]);
            put = new Put(rk);
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("score"), Bytes.toBytes(scoresInfo[2]));
            scoresPuts.add(put);
        }

        scores.put(scoresPuts);
    }

    public static void putStudentsInfo(Connection conn, BufferedReader bufferedReader) throws Exception {
        //获取表的实例
        Table students = MyHBaseTools.getTableExample(conn, "students");
        //创建学生表数据put集合对象
        ArrayList<Put> studentPuts = new ArrayList<>();

        //将学生数据添加到hbase表中，以学号作为rowkey
        String line = null;
        Put put = null;
        while ((line = bufferedReader.readLine()) != null) {
            String[] studentInfo = line.split(",");
            byte[] id = Bytes.toBytes(studentInfo[0]);
            put = new Put(id);
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(studentInfo[1]));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(studentInfo[2]));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes(studentInfo[3]));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("clazz"), Bytes.toBytes(studentInfo[4]));
            studentPuts.add(put);
        }
        students.put(studentPuts);
    }
}
